package org.jetlinks.pro.things.impl;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.Configurable;
import org.jetlinks.core.codec.Codec;
import org.jetlinks.core.codec.Codecs;
import org.jetlinks.core.event.EventBus;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.things.*;
import org.jetlinks.core.utils.StringBuilderUtils;
import org.jetlinks.pro.PropertyConstants;
import org.jetlinks.pro.things.*;
import org.jetlinks.pro.utils.TopicUtils;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;


/**
 * 物连接器管理器,用于管理连接器,并从连接器中获取数据进行持久化等处理.
 *
 * @author zhouhao
 * @see ThingsDataRepository
 * @since 1.12
 */
@AllArgsConstructor
@Slf4j
public class DefaultThingsConnectorManager implements ThingsConnectorManager {

    private final static List<String> allConfigHeader = Arrays.asList(
        ThingsConfigKeys.templateId.getKey(),
        PropertyConstants.tenantId.getKey(),
        PropertyConstants.groupId.getKey(),
        PropertyConstants.tenantMemberId.getKey(),
        PropertyConstants.orgId.getKey(),
        PropertyConstants.bindings.getKey()
    );

    //消息编码器，用于消息总线发布消息时对消息进行编码
    private final static Codec<ThingMessage> MESSAGE_CODEC = Codecs.lookup(ThingMessage.class);

    //以物ID为key的连接器缓存
    private final Map<ThingId, Map<MetadataId, ThingsConnector>> connectors = new ConcurrentHashMap<>();

    //消息总线,消息的订阅发布实现
    private final EventBus eventBus;

    //物注册中心,统一管理物的基础信息以及配置等信息
    private final ThingsRegistry thingsRegistry;

    /**
     * 给指定到物注册一个连接器,表示将连接器中的数据绑定到指定到物实例,可调用返回值{@link Disposable#dispose()}来注销此连接器
     *
     * @param thingId   物ID
     * @param connector 连接器
     * @return Disposable
     */
    @Nonnull
    @Override
    public Disposable register(@Nonnull ThingId thingId,
                               @Nonnull MetadataId metadataId,
                               @Nonnull ThingsConnector connector,
                               boolean template) {

        Map<MetadataId, ThingsConnector> cache = connectors
            .computeIfAbsent(thingId, id -> new ConcurrentHashMap<>());

        ThingsConnector thingsConnector = cache
            .compute(metadataId, (id, old) -> {
                if (old != null) {
                    old.dispose();
                }
                // FIXME: 2021/11/4 支持热替换？
                return connector;
            });
        if (template) {
            return thingsConnector.connectTo(thingId, metadataId, this::handleThingMessage);
        }
        return thingsConnector.connectTo(metadataId, this::handleThingMessage);

    }

    /**
     * 获取物的连接器
     *
     * @param thingId 物ID
     * @return ThingsConnector
     */
    @Override
    public Mono<ThingsConnector> getConnector(@Nonnull ThingId thingId, @Nonnull MetadataId metadataId) {
        Map<MetadataId, ThingsConnector> cached = connectors.get(thingId);
        if (null == cached) {
            return Mono.empty();
        }
        return Mono.justOrEmpty(cached.get(metadataId));
    }

    /**
     * 移除连接器,如果物模型ID为null,则移除全部
     *
     * @param thingId    物ID
     * @param metadataId 物模型ID
     * @return void
     */
    @Override
    public Mono<Void> removeConnector(@Nonnull ThingId thingId,
                                      @Nullable MetadataId metadataId) {
        Map<MetadataId, ThingsConnector> cached = connectors.get(thingId);
        if (null == cached) {
            return Mono.empty();
        }
        if (null != metadataId) {
            ThingsConnector connector = cached.remove(metadataId);
            //移除连接器时释放连接器资源
            if (null != connector) {
                connector.dispose();
            }
        } else {
            cached = connectors.remove(thingId);
            //移除连接器时释放连接器资源
            if (null != cached) {
                cached.values().forEach(Disposable::dispose);
            }
        }
        return Mono.empty();
    }

    /**
     * 处理物消息，将连接器处理后的消息从新推送到消息总线
     *
     * @param message 连接器处理后的物消息
     */
    public Mono<Void> handleThingMessage(ThingMessage message) {
        ThingType thingType = ThingType.of(message.getThingType());

        return thingsRegistry
            .getThing(thingType, message.getThingId())
            .flatMap(thing -> refactorMessage(message, thing))
            .flatMap(msg -> Flux
                .fromIterable(createMessageTopic(msg))
                .flatMap(topic -> eventBus.publish(topic, MESSAGE_CODEC, msg))
                .then()
            ).onErrorResume(err -> {
                log.error(err.getMessage(), err);
                return Mono.empty();
            });
    }

    /**
     * 创建推送到物消息到消息总线的topic
     *
     * @param message 连接器处理后的物消息
     */
    public static Set<String> createMessageTopic(ThingMessage message) {

        String topic = StringBuilderUtils
            .buildString(message, (msg, builder) -> {
                // 为了兼容设备消息(device),使用/{type}/{templateId}/{thingId}/.....
                builder
                    .append("/")
                    .append(msg.getThingType())
                    .append("/")
                    .append(msg.getHeaderOrElse(ThingsConfigKeys.templateId.getKey(), () -> "null"))
                    .append("/")
                    .append(msg.getThingId());
                TopicUtils.appendMessageTopic(msg, builder);
            });

        return TopicUtils.refactorTopic(message, topic);
    }

    /**
     * 重构物消息，将物中的配置信息放入消息header中
     * @param message 物消息
     * @param configurable 物配置
     */
    public static Mono<ThingMessage> refactorMessage(ThingMessage message, Configurable configurable) {
        return configurable
            .getConfigs(allConfigHeader)
            .map(values -> {
                values.getAllValues().forEach(message::addHeader);
                return message;
            });
    }


}
